package com.hucais.utils;

import com.hucais.app.ods.SelectMonitorApp;
import com.hucais.bean.TransientSink;
import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcExecutionOptions;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.connector.jdbc.JdbcStatementBuilder;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.lang.reflect.Field;

public class JdbcUtil {
    private static final Logger LOG = LoggerFactory.getLogger(SelectMonitorApp.class);
    // 批量写入数据的大小
    private static final int BATCH_SIZE = 500;
    // 每次批量写入最小间隔时间
    private static final int INTERVAL_MILLIS = 30 * 1000;

    public static <T> SinkFunction<T> getMysqlSinkFunc(String sql) {
        // 初始化配置
        String DB_URL = DefaultPropertiesUtil.get("mysql.db_url");
        String JDBC_DRIVER = DefaultPropertiesUtil.get("mysql.jdbc_driver");
        String USER = DefaultPropertiesUtil.get("mysql.user");
        String PASS = DefaultPropertiesUtil.get("mysql.pass");

        JdbcConnectionOptions connectionOptions = new JdbcConnectionOptions.JdbcConnectionOptionsBuilder().withDriverName(JDBC_DRIVER).withUrl(DB_URL)
                .withUsername(USER).withPassword(PASS).build();
        return initRdbmsSink(connectionOptions, sql);
    }

    private static <T> SinkFunction<T> initRdbmsSink(JdbcConnectionOptions connectionOptions, String sql) {
        JdbcExecutionOptions executionOptions = new JdbcExecutionOptions.Builder().withBatchSize(BATCH_SIZE).withBatchIntervalMs(INTERVAL_MILLIS).build();

        return JdbcSink.sink(sql, (JdbcStatementBuilder<T>) (preparedStatement, t) -> {
                    //通过反射的方式获取数据中的内容
                    //1.获取所有的列信息(包含私有属性)
                    Field[] declaredFields = t.getClass().getDeclaredFields();

                    //2.遍历列信息
                    int offset = 0;
                    for (int i = 0; i < declaredFields.length; i++) {
                        Field field = declaredFields[i];
                        //设置私有属性信息可获取
                        field.setAccessible(true);
                        //获取字段上的注解信息
                        TransientSink transientSink = field.getAnnotation(TransientSink.class);
                        if (transientSink != null) {
                            offset++;
                            continue;
                        }
                        try {
                            Object o = field.get(t);
                            preparedStatement.setObject(i + 1 - offset, o);
                        } catch (IllegalAccessException e) {
                            LOG.error("拼接SQL出错!", e);
                        }
                    }
                },
                executionOptions, connectionOptions
        );
    }
}
